package main.scala.com.lagou.spark.part02

import org.apache.log4j.{Level, Logger}
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author: yehw
 * @date: 2020/11/12 21:51
 * @description: 第二题
 */
object part02 {
  def main(args: Array[String]): Unit = {
    // 初始化
    Logger.getLogger("org").setLevel(Level.ERROR)
    val conf: SparkConf = new SparkConf()
      .setAppName(this.getClass.getCanonicalName)
      .setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("warn")
    //定义顶点
    //1, "SFO"
    //2, "ORD"
    //3, "DFW"
    val vertexArray: Array[(VertexId, (String))] = Array(
      (1L, ("SFO")),
      (2L, ("ORD")),
      (3L, ("DFW")))
    //定义边
    //1, 2,1800
    //2, 3, 800
    //3, 1, 1400
    val edgeArray: Array[Edge[Int]] = Array(
      Edge(1L, 2L, 1800),
      Edge(2L, 3L, 800),
      Edge(3L, 1L, 1400))
    //定义构造vertexRDD和edgeRDD
    val vertexRDD: RDD[(Long, (String))] = sc.makeRDD(vertexArray)
    val edgeRDD: RDD[Edge[Int]] = sc.makeRDD(edgeArray)
    // 构造图Graph[VD,ED]
    val graph: Graph[(String), Int] = Graph(vertexRDD, edgeRDD)
    //请所有的定点
    println("**********所有的顶点*************")
    graph.vertices.foreach(println)
    println("**********所有的边*************")
    graph.edges.foreach(println)
    println("**********所有的triplets*************")
    graph.triplets.foreach(println)
    println("**********顶点数*************")
    println(graph.vertices.count())
    println("**********顶边数*************")
    println(graph.edges.count())
    println("**********机场距离大于1000的有几个，有哪些*************")
    println(graph.edges.filter(edge=>edge.attr>1000).count())
    graph.edges.filter(edge => edge.attr > 1000).foreach(println)
    println("**********按所有机场之间的距离排序（降序），输出结果*************")
    graph.connectedComponents.edges.sortBy{x=>{x.attr}}.foreach(print)
  }
}
